package cgl.granules.application;

import java.io.IOException;
import java.util.Properties;

import cgl.granules.dataset.DatasetException;
import cgl.granules.dataset.StreamingAccess;
import cgl.granules.exception.MarshallingException;
import cgl.granules.operation.MapReduceBase;
import cgl.granules.operation.MapReduceException;
import cgl.granules.results.ResultPayload;
import cgl.granules.results.Results;
import cgl.granules.results.ResultsFactory;

import cgl.narada.event.NBEvent;
import cgl.narada.event.TemplateProfileAndSynopsisTypes;
import cgl.narada.service.*;
import cgl.narada.service.client.*;

/***
 * 
 * A reduce to collect ASCII and POS data from the maps it subscribes to and
 * publish them to NaradaBrokering.
 * 
 * @author Chao Sun
 * 
 */

public class RyoReduce extends MapReduceBase {

	private String publishString[] = new String[2];
	private StreamingAccess mapReduceDataset = null;
	private ResultPayload resultPayload = null;
	private ClientService clientService = null;
	private String publishTopic = null;

	public RyoReduce() {
		setDomain("Generic");
		setVersionInformation(200);
		setOperationIdentifier("Ryo-Reduce");
		publishString[0] = null;
		publishString[1] = null;
		resultPayload = ResultPayload.getInstance();
	}

	public void publishData(byte[] publishBytes) throws ServiceException,
			MapReduceException, MarshallingException, InterruptedException {
		if (publishTopic == null) {
			Properties props = getProcessingDirectives();
			if (props == null) {
				System.out.println("no property in reduce! exiting...");
				return;
			}
			if (!props.containsKey("pubTopic")) {
				System.out.println("incorrect property in reduce! exiting...");
				return;
			}
			publishTopic = props.getProperty("pubTopic");
		}
		
		Results results = createResults(true, true);
		results.setResultPayload(publishBytes);
		writeResults(publishTopic, results);
	}

	public void execute() {
		try {
			mapReduceDataset = getMapReduceDataset();

			while (mapReduceDataset.isDataAvailable()) {
				NBEvent nbEvent = mapReduceDataset.getStreamingData();
				processResultsFromMap(nbEvent);
				nbEvent = null;
			}

		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	private synchronized void processResultsFromMap(NBEvent nbEvent)
			throws IOException, MarshallingException, MapReduceException,
			ServiceException, InterruptedException {
		byte[] resultBytes = nbEvent.getContentPayload();
		Results results = ResultsFactory.getInstance().getResults(resultBytes);
		byte[] reportBytes = results.getResultPayload();

		try {
			String dataString = resultPayload.getString(reportBytes);
			byte[] dataBytes = dataString.getBytes();

			if (dataString.startsWith("ASCII")) {
				publishString[0] = new String(dataBytes, 5,
						dataBytes.length - 5);
			} else if (dataString.startsWith("POS")) {
				publishString[1] = new String(dataBytes, 3,
						dataBytes.length - 3);
			} else {
				System.out.println("Incorrect data! exiting ...");
				return;
			} // [0] for ASCII, [1] for POS

			dataString = null;
			dataBytes = null;
		} catch (Exception e) {
			e.printStackTrace();
		}

		if (publishString[0] != null && publishString[1] != null) {
			// System.out.println("ASCII:\n" + publishString[0]);
			// System.out.println("POS:\n" + publishString[1]);
			byte[] publishBytes = resultPayload.getBytes(publishString);
			publishData(publishBytes);

			publishString[0] = null;
			publishString[1] = null;
		}

		resultBytes = null;
		results = null;
		reportBytes = null;
	}

}